/* 
 * Copyright (c), nFractals Ltd. All Rights Reserved.
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at 
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package org.parade.driver.hazelcast;

import java.util.Date;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;

import org.parade.distributed.Condition;
import org.parade.distributed.Id;

import com.hazelcast.core.AtomicNumber;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IQueue;
import com.hazelcast.core.IdGenerator;
import com.hazelcast.core.ItemListener;

public class Driver extends org.parade.driver.Driver{
	
	HazelcastInstance hazelCast;
	
	public Driver(){
		hazelCast = Hazelcast.getDefaultInstance();
	}
	
	public Driver(HazelcastInstance hazelCast){
		this.hazelCast = hazelCast;
	}
	
	
	@Override
	public Lock getLock(Id id) {
		return hazelCast.getLock(id.toString());
	}
	
	public long generateUniqueId(){
		IdGenerator idgen = hazelCast.getIdGenerator("IDGEN");
		return idgen.newId();
	}

	@Override
	public java.util.concurrent.locks.Condition getCondition(
			org.parade.distributed.Lock lock) {
		return new Condition(lock);
	}
	


	private class Condition implements java.util.concurrent.locks.Condition, ItemListener<Object> {
		
		private IQueue<Object> queue;
		private org.parade.distributed.Lock lock;
		private Long innerSyncObject = new Long(0);

		public Condition(org.parade.distributed.Lock lock){
			this.lock = lock;
			queue = hazelCast.getQueue(lock.getDistributedObjectId().toString()+"-CONDITION-QUEUE");
			queue.addItemListener(this, true);
		}
		
		@Override
		public void await() throws InterruptedException {
			synchronized(innerSyncObject){
				try{
					lock.unlock();
					innerSyncObject.wait();
				} catch(Exception e){
					throw new InterruptedException();
				} finally{
					lock.lock();
				}
			}
		}

		@Override
		public boolean await(long time, TimeUnit unit) throws InterruptedException {
			return (awaitNanos(unit.toNanos(time)) > 0);
		}
		
		
		@Override
		public long awaitNanos(long nanosTimeout) throws InterruptedException {
			synchronized(innerSyncObject){
			
				try{
					lock.unlock();	
					long milli = nanosTimeout / 1000000;
					int nano = (int) (nanosTimeout % 1000000);
					long startTime = System.currentTimeMillis()*100000;
					innerSyncObject.wait(milli, nano);
					long endTime = System.currentTimeMillis()*1000000;
					long diff = nanosTimeout - (endTime - startTime);
					return diff;
				} catch(Exception e){
					throw new InterruptedException();
				} finally{
					lock.lock();
				}
			}
		}

		@Override
		public void awaitUninterruptibly() {
			synchronized(innerSyncObject){
				try{
					lock.unlock();
				
					while(true){
						try{
							innerSyncObject.wait();
							break;
						} catch(InterruptedException e){
							continue;
						}
					}
				} catch(IllegalMonitorStateException e){
					throw e;
				} catch(Exception e){
					e.printStackTrace();
				} finally{
					lock.lock();
				}
			}
		}

		@Override
		public boolean awaitUntil(Date deadline) throws InterruptedException {
			long timeout = deadline.getTime();
			if(timeout <= 0){
				return true;
			} else {
				return await(timeout, TimeUnit.MILLISECONDS);
			}
		}

		@Override
		public void signal() {
			try{
				synchronized(queue){
					queue.add(new Long(0));
					queue.clear();
				}
			} catch(Exception e){
		
			} finally{
			}
		}

		@Override
		public void signalAll() {
			try{
				synchronized(queue){
					queue.add(new Long(1));
					queue.clear();
				}
			} catch(Exception e){
				
			} finally{
			}
		}
		

		@Override
		public void itemAdded(Object arg0) {
			synchronized(innerSyncObject){
				Long value = (Long) arg0;
				if(value==0){
					innerSyncObject.notify();
				} else {
					innerSyncObject.notifyAll();
				}
			}
		}

		@Override
		public void itemRemoved(Object arg0) {
			
		}

	}


	@SuppressWarnings("rawtypes")
	@Override
	public BlockingQueue getBlockingQueue(Id id) {
		return hazelCast.getQueue(id.toString());
	}

	@Override
	public Map getMap(Id id) {
		return hazelCast.getMap(id.toString());
	}
	
}
